package com.cn.client;

import com.cn.client.impl.DefaultMQPushConsumerImpl;
import com.cn.common.*;
import com.cn.common.processor.ProcessorCode;
import com.cn.msg.Message;
import com.cn.producer.Producer;
import com.cn.util.RemotingSerializable;
import com.cn.util.UtilAll;

import java.util.*;
import java.util.concurrent.*;

public class MQClientInstance {

    private final PullMessageService pullMessageService;

    /**
     * 存储消息生成端
     */
    private ConcurrentMap<String /* producerGroup */ , Producer> producerTables = new ConcurrentHashMap<>();
    /**
     * 存储消息消费端
     */
    private ConcurrentMap<String /* consumeGroup */, DefaultMQPushConsumerImpl> consumerConcurrentMap = new ConcurrentHashMap<>();
    private MqClientApiImpl mqClientApi = null;
    private DefaultMQPushConsumerImpl defaultMQPushConsumer;
    private NettyClientConfig nettyClientConfig;
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    private ConcurrentMap<String /*topic*/, TopicRouteData> topicRouteTables = new ConcurrentHashMap<>();
    private String clientId;


    public MQClientInstance(DefaultMQPushConsumerImpl defaultMQPushConsumer) {
        this.defaultMQPushConsumer = defaultMQPushConsumer;
        this.pullMessageService = new PullMessageService(this);
        this.nettyClientConfig = new NettyClientConfig();
        this.mqClientApi = new MqClientApiImpl(this.nettyClientConfig);
        this.clientId = UtilAll.getPid() + "";
    }

    public void start() {
        this.mqClientApi.start();
        //拉取消息服务
        this.pullMessageService.start();

        //发送心跳包
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                sendHeartbeat();
            }
        }, 1, 5, TimeUnit.SECONDS);//消费端的详细信息

        //更新topic
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                defaultMQPushConsumer.updateTopicChanged();
            }
        }, 10, 60, TimeUnit.SECONDS);//每分钟执行一次

        //持久化消费位移
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                defaultMQPushConsumer.getOffsetStore().persitAll();
            }
        }, 10, 5, TimeUnit.SECONDS);//每五秒执行一次
    }

    public void shutdown() {
        this.mqClientApi.shutdown();
    }

    //发送心跳
    public void sendHeartbeat() {
        try {
            ClientInfo clientInfo = new ClientInfo();
            clientInfo.setGroup(this.defaultMQPushConsumer.getConsumeGroup());
            clientInfo.setCientId(this.getClientId());
            this.mqClientApi.sendHeartbeat(clientInfo);
//            System.out.println("发送消费端注册信息");
        } catch (Exception e) {
            System.out.println("注册消费信息异常：" + e.getMessage());
//            e.printStackTrace();
        }
    }

    public DefaultMQPushConsumerImpl selectDefaultMQPushConsumer(String group) {
        return this.consumerConcurrentMap.get(group);
    }

    public void register(String consumeGroup, DefaultMQPushConsumerImpl defaultMQPushConsumer) {
        this.consumerConcurrentMap.putIfAbsent(consumeGroup, defaultMQPushConsumer);
    }

    public void unregister(String group) {
        this.consumerConcurrentMap.remove(group);
    }

    public MqClientApiImpl getMqClientApi() {
        return mqClientApi;
    }

    public void updateTopic(String topic) {
        byte[] bytes = this.getMqClientApi().updateTopicRouteData(topic);
        if (bytes != null) {
            TopicRouteData topicRouteData = RemotingSerializable.decode(bytes, TopicRouteData.class);
            if (topicRouteData != null) {
                if (topicRouteData.getBrokerData().isEmpty()) {
                    return;
                }
                for (BrokerData brokerData : topicRouteData.getBrokerData()) {
                    this.getMqClientApi().getBrokerAddrTable().put(brokerData.getBrokerName(), brokerData.getAddr());
                }

                boolean changed = false;
                TopicRouteData old = this.topicRouteTables.get(topic);
                if (old == null) {
                    changed = true;
                } else {
                    if (!old.equals(topicRouteData)) {
                        changed = true;
                    }
                }

                //订阅的消息存在改变
                if (changed) {
                    Set<MessageQueue> messageQueueSet = updateMessageQueue(topicRouteData);
                    this.defaultMQPushConsumer.reshMessageQueue(topic, messageQueueSet);
                    this.topicRouteTables.put(topic, topicRouteData);

                    //producer
                    {
                        Iterator<Map.Entry<String, Producer>> producerIterator = this.producerTables.entrySet().iterator();
//                        while (producerIterator.hasNext()) {
//                            Map.Entry<String, Producer> entry = producerIterator.next();
//                            Producer producer = entry.getValue();
//                            if (producer != null) {
//                                producer.updateTopicPublishInfo(topic,null);
//                            }
//                        }
                    }
                }
            }
        }
    }

    public Set<MessageQueue> updateMessageQueue(TopicRouteData topicRouteData) {
        Set<MessageQueue> messageQueueSet = new HashSet<>();
        List<QueueData> queueDataList = topicRouteData.getQueueData();
        for (QueueData queueData : queueDataList) {
            for (int i = 0; i < queueData.getReadQueue(); i++) {
                MessageQueue messageQueue = new MessageQueue();
                messageQueue.setTopic(queueData.getTopic());
                messageQueue.setQueueId(i);
                messageQueue.setBrokerName(queueData.getBrokerName());
                messageQueueSet.add(messageQueue);
            }
        }
        return messageQueueSet;
    }

    public List<String> pullClientList(String topic) {
        TopicRouteData topicRouteData = topicRouteTables.get(topic);
        if (topicRouteData != null) {
            if (topicRouteData.getBrokerData() != null) {
                //去取第一个节点
                String addr = topicRouteData.getBrokerData().get(0).getAddr().get(0);
                try {
                    return this.mqClientApi.pullClientList(this.defaultMQPushConsumer.getConsumeGroup(), addr);
                } catch (Exception e) {
//                    e.printStackTrace();
                }
            }
        }
        return null;
    }

    public void updateTopic(TopicPublishInfo topicPublishInfo) {

    }

    public String getClientId() {
        return clientId;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    public PullMessageService getPullMessageService() {
        return pullMessageService;
    }

    public void registerProducer(String group, Producer producer) {
        this.producerTables.putIfAbsent(group, producer);
    }
}
